package com.guqj.mq.kafka;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class Consumer {

	private KafkaConsumer<String, String> consumer = null;

	public Consumer() {
		Properties props = new Properties();

		props.put("bootstrap.servers", KafkaConstant.KAFKA_ADDRESS);
		// group 代表一个消费组
		props.put("group.id", KafkaConstant.KAFKA_GROUP_NAME);
		props.put("enable.auto.commit", "true");

		// zookeeper 配置
		props.put("zookeeper.connect", KafkaConstant.ZOOKEEPER_ADDRESS);
		props.put("zookeeper.session.timeout.ms", "10000");
		props.put("zookeeper.sync.time.ms", "500");
		props.put("auto.commit.interval.ms", "1000");

		// 序列化
		props.put("value.deserializer", KafkaConstant.KAFKA_SERIALIZER_CLASS);
		props.put("key.deserializer", KafkaConstant.KAFKA_SERIALIZER_CLASS);

		consumer = new KafkaConsumer<String, String>(props);
		consumer.subscribe(Arrays.asList(KafkaConstant.KAFKA_TOPIC));
	}

	public void consume() {
		while (true) {
			ConsumerRecords<String, String> records = consumer.poll(100);
			for (ConsumerRecord<String, String> record : records)
				System.out
						.printf("offset = %d, key = %s, value = %s \n", record.offset(), record.key(), record.value());
		}
	}

	public static void main(String[] args) {
		new Consumer().consume();
	}
}
